package com.hps.demo.msg;

import com.alibaba.fastjson.JSON;
import com.hps.demo.entity.User;
import com.hps.demo.service.IUserService;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author heps
 * @since 2020/11/13 15:36
 */
@Slf4j
@Component
public class UserTopicListener implements InitializingBean, DisposableBean {

  private static final BlockingQueue<User> USER_QUEUE = new LinkedBlockingQueue<>();
  
  private volatile Boolean running = true;

  @Resource
  private IUserService userService;

  @Resource
  private Executor bizExecutor;

  @KafkaListener(topics = "userTopic", groupId = "userTopicGroup")
  public void listen(ConsumerRecord<String, String> msg) {
    User user = JSON.parseObject(msg.value(), User.class);
    USER_QUEUE.offer(user);
    if (USER_QUEUE.size() > 3000) {
      try {
        TimeUnit.SECONDS.sleep(1);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public void saveData() {
    List<User> users = new ArrayList<>();
    while (running) {
      try {
        User user = USER_QUEUE.poll(10, TimeUnit.MILLISECONDS);
        if (Objects.nonNull(user)) {
          users.add(user);
        } else {
          TimeUnit.MILLISECONDS.sleep(500);
        }
        if (users.size() > 0 && (Objects.isNull(user) || users.size() % 100 == 0)) {
          log.info("user queue size: {}", USER_QUEUE.size());
          userService.saveBatch(users);
          users.clear();
        }
      } catch (InterruptedException e) {
        log.error(e.getMessage(), e);
      }

    }
    while (USER_QUEUE.size() > 0) {
      users.add(USER_QUEUE.poll());
    }
    if (users.size() > 0) {
      userService.saveBatch(users);
    }
    try {
      TimeUnit.SECONDS.sleep(5);
      if (USER_QUEUE.size() > 0) {
        log.info("the user queue is not empty, good consumer cannot stop");
        startConsumer();
      } else {
        log.info("user consumer stopped...");
      }
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public void setRunning(boolean running) {
    this.running = running;
  }

  public void startConsumer() {
    if (!running) {
      synchronized (running) {
        if (!running) {
          running = true;
          bizExecutor.execute(this::saveData);
        }
      }
    }
  }

  @Override
  public void afterPropertiesSet() throws Exception {
    bizExecutor.execute(this::saveData);
    log.info("user topic listener started...");
  }

  @Override
  public void destroy() throws Exception {
    if (running) {
      running = false;
      TimeUnit.SECONDS.sleep(5);
    }
  }
}
